/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.broker.client.rebalance;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.message.MessageQueue;

/**
 * 顺序消息争抢队列锁
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-6-26
 */
public class RebalanceLockManager {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.RebalanceLockLoggerName);
	private final static long RebalanceLockMaxLiveTime = Long.parseLong(System.getProperty("rocketmq.broker.rebalance.lockMaxLiveTime", "60000"));
	private final Lock lock = new ReentrantLock();
	private final ConcurrentHashMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>> mqLockTable = new ConcurrentHashMap<String, ConcurrentHashMap<MessageQueue, LockEntry>>(1024);

	class LockEntry {
		private String clientId;
		private volatile long lastUpdateTimestamp = System.currentTimeMillis();

		public String getClientId() {
			return clientId;
		}

		public void setClientId(String clientId) {
			this.clientId = clientId;
		}

		public long getLastUpdateTimestamp() {
			return lastUpdateTimestamp;
		}

		public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
			this.lastUpdateTimestamp = lastUpdateTimestamp;
		}

		public boolean isExpired() {
			boolean expired = (System.currentTimeMillis() - this.lastUpdateTimestamp) > RebalanceLockMaxLiveTime;

			return expired;
		}

		public boolean isLocked(final String clientId) {
			boolean eq = this.clientId.equals(clientId);
			return eq && !this.isExpired();
		}
	}

	private boolean isLocked(final String group, final MessageQueue mq, final String clientId) {
		// 以group从RebalanceLockManager.mqLockTable中获取该consumerGroup下面的所有ConcurrentHashMap<MessageQueue, LockEntry>集合
		ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
		if (groupValue != null) {
			// 若该集合不为空再以MessageQueue对象获取对应的LockEntry标记类
			LockEntry lockEntry = groupValue.get(mq);
			if (lockEntry != null) {
				boolean locked = lockEntry.isLocked(clientId);
				if (locked) {
					// 若是被该clientId锁住且没有过期，则更新LockEntry标记类的lastUpdateTimestamp变量
					lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
				}

				return locked;
			}
		}
		// 若该集合为空则认为没有锁住
		return false;
	}

	/**
	 * 尝试锁队列
	 * 
	 * @return 是否lock成功
	 */
	public boolean tryLock(final String group, final MessageQueue mq, final String clientId) {
		// 没有被锁住
		if (!this.isLocked(group, mq, clientId)) {
			try {
				this.lock.lockInterruptibly();
				try {
					ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
					if (null == groupValue) {
						groupValue = new ConcurrentHashMap<MessageQueue, LockEntry>(32);
						this.mqLockTable.put(group, groupValue);
					}

					LockEntry lockEntry = groupValue.get(mq);
					if (null == lockEntry) {
						lockEntry = new LockEntry();
						lockEntry.setClientId(clientId);
						groupValue.put(mq, lockEntry);
						log.info("tryLock, message queue not locked, I got it. Group: {} NewClientId: {} {}", //
								group, //
								clientId, //
								mq);
					}

					if (lockEntry.isLocked(clientId)) {
						lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
						return true;
					}

					String oldClientId = lockEntry.getClientId();

					// 锁已经过期，抢占它
					if (lockEntry.isExpired()) {
						lockEntry.setClientId(clientId);
						lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
						log.warn("tryLock, message queue lock expired, I got it. Group: {} OldClientId: {} NewClientId: {} {}", //
								group, //
								oldClientId, //
								clientId, //
								mq);
						return true;
					}

					// 锁被别的Client占用
					log.warn("tryLock, message queue locked by other client. Group: {} OtherClientId: {} NewClientId: {} {}", //
							group, //
							oldClientId, //
							clientId, //
							mq);
					return false;
				} finally {
					this.lock.unlock();
				}
			} catch (InterruptedException e) {
				log.error("putMessage exception", e);
			}
		}
		// 已经锁住，尝试更新时间
		else {
			// isLocked 中已经更新了时间，这里不需要再更新
		}

		return true;
	}

	/**
	 * 批量方式锁队列，返回锁定成功的队列集合
	 * 
	 * @return 是否lock成功
	 */
	public Set<MessageQueue> tryLockBatch(final String group, final Set<MessageQueue> mqs, final String clientId) {
		// 存储锁住的MessageQueue集合
		Set<MessageQueue> lockedMqs = new HashSet<MessageQueue>(mqs.size());
		// 未锁住的MessageQueue集合；
		Set<MessageQueue> notLockedMqs = new HashSet<MessageQueue>(mqs.size());

		// 先通过不加锁的方式尝试查看哪些锁定，哪些没锁定
		for (MessageQueue mq : mqs) {
			// 检查该MessageQueue对象是否被锁住
			if (this.isLocked(group, mq, clientId)) {
				// 若遍历的MessageQueue对象被该ClientId锁住且锁未超期了则将MessageQueue对象存入lockedMqs变量中
				lockedMqs.add(mq);
			} else {
				// 否则将MessageQueue对象存入notLockedMqs变量中。
				notLockedMqs.add(mq);
			}
		}

		// 若notLockedMqs集合不为空，即表示有未锁住的MessageQueue队列
		// ==============下面的逻辑处理块是加锁的，就是说同时只有一个线程执行该逻辑块。===============
		if (!notLockedMqs.isEmpty()) {
			try {
				this.lock.lockInterruptibly();
				try {
					// RebalanceLockManager.mqLockTable:ConcurrentHashMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>>，
					// 表示一个consumerGroup下面的所有MessageQueue的锁的情况
					/**
					 * 锁是由LockEntry类标记的，
					 * 在LockEntry类中clientId和lastUpdateTimestamp两个变量，
					 * 表示某个客户端（clientId）在某时间（lastUpdateTimestamp）对MessageQueue进行锁住，锁的超时时间为60秒；
					 */
					// 以group从RebalanceLockManager.mqLockTable中获取该consumerGroup下面的所有ConcurrentHashMap<MessageQueue, LockEntry>集合取名groupvalue集合
					ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
					if (null == groupValue) {
						// 4、若该集合为空，则新建一个ConcurrentHashMap<MessageQueue,LockEntry>集合并赋值给groupvalue集合，
						groupValue = new ConcurrentHashMap<MessageQueue, LockEntry>(32);
						// 然后以consumerGroup为key值存入RebalanceLockManager.mqLockTable变量中；
						this.mqLockTable.put(group, groupValue);
					}

					// 遍历没有锁住的队列
					for (MessageQueue mq : notLockedMqs) {
						// 以每个MessageQueue对象从groupvalue集合中取LockEntry标记类
						LockEntry lockEntry = groupValue.get(mq);
						if (null == lockEntry) {
							// 若该类为null（在第4步新建的集合会出现此情况）则初始化LockEntry类
							lockEntry = new LockEntry();
							// 设置该类的clientId为该Consumer的clientId
							lockEntry.setClientId(clientId);
							groupValue.put(mq, lockEntry);
							log.info("tryLockBatch, message queue not locked, I got it. Group: {} NewClientId: {} {}", //
									group, //
									clientId, //
									mq);
						}

						// 已经锁定
						// 若lockEntry不为null，则再次检查是否被该clientId锁住
						if (lockEntry.isLocked(clientId)) {
							// 若被该clientId锁住则更新LockEntry标记类的lastUpdateTimestamp变量，
							lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
							// 并添加到lockedMqs集合中，
							lockedMqs.add(mq);
							// 继续遍历notLockedMqs集合中的下一个MessageQueue对象
							continue;
						}

						String oldClientId = lockEntry.getClientId();

						// 锁已经过期，抢占它
						// 若仍然未被该clientId锁住锁则检查锁是否超时
						if (lockEntry.isExpired()) {
							// 若已经超时则设置该LockEntry类的clientId为该Consumer的clientId
							lockEntry.setClientId(clientId);
							// 并更新lastUpdateTimestamp变量；
							lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
							log.warn("tryLockBatch, message queue lock expired, I got it. Group: {} OldClientId: {} NewClientId: {} {}", //
									group, //
									oldClientId, //
									clientId, //
									mq);
							lockedMqs.add(mq);
							continue;
						}

						// 锁被别的Client占用
						log.warn("tryLockBatch, message queue locked by other client. Group: {} OtherClientId: {} NewClientId: {} {}", //
								group, //
								oldClientId, //
								clientId, //
								mq);
					}
				} finally {
					this.lock.unlock();
				}
			} catch (InterruptedException e) {
				log.error("putMessage exception", e);
			}
		}

		return lockedMqs;
	}

	public void unlockBatch(final String group, final Set<MessageQueue> mqs, final String clientId) {
		try {
			this.lock.lockInterruptibly();
			try {
				// 以consumerGroup为key值从
				// RebalanceLockManager.mqLockTable:ConcurrentHashMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>>变量中
				// 获取对应的ConcurrentHashMap<MessageQueue, LockEntry>集合；
				ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
				if (null != groupValue) {
					// 遍历入参Set<MessageQueue>集合
					for (MessageQueue mq : mqs) {
						// 以该集合中的每个MessageQueue对象从上一步获得的ConcurrentHashMap<MessageQueue, LockEntry>集合中获取LockEntry标记类
						LockEntry lockEntry = groupValue.get(mq);
						if (null != lockEntry) {
							// 若该类不为null，则检查该LockEntry类的ClientId是否等于请求中的ClientId（请求端的ClientId）
							if (lockEntry.getClientId().equals(clientId)) {
								// 若相等则将该MessageQueue对象的记录从ConcurrentHashMap<MessageQueue, LockEntry>集合中移除；其他情况均不处理；
								groupValue.remove(mq);
								log.info("unlockBatch, Group: {} {} {}", //
										group, //
										mq, //
										clientId);
							} else {
								log.warn("unlockBatch, but mq locked by other client: {}, Group: {} {} {}", //
										lockEntry.getClientId(), //
										group, //
										mq, //
										clientId);
							}
						} else {
							log.warn("unlockBatch, but mq not locked, Group: {} {} {}", //
									group, //
									mq, //
									clientId);
						}
					}
				} else {
					log.warn("unlockBatch, group not exist, Group: {} {}", //
							group, //
							clientId);
				}
			} finally {
				this.lock.unlock();
			}
		} catch (InterruptedException e) {
			log.error("putMessage exception", e);
		}
	}
}
